package org.apache.lorry.demo.producer.interceptor;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;
@Slf4j
public class SendLogProducerInterceptor implements ProducerInterceptor<String,String> {
    //调用send()后，消息序列化和计算分区之前
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        log.info("current Thread :{} SendLogProducerInterceptor onSend: topic: {}, key: {}, value: {}", Thread.currentThread(), record.topic(), record.key(), record.value());
        return record;
    }
    //消息被应答之前或消息发送失败时
    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            log.error("current Thread :{}  SendLogProducerInterceptor onAcknowledgement error: ", Thread.currentThread(),exception);
        }else{
            log.info("current Thread :{}  SendLogProducerInterceptor onAcknowledgement: topic: {}, partition: {}, offset: {}", Thread.currentThread(), metadata.topic(), metadata.partition(), metadata.offset());
        }
    }

//    生产者关闭时
    @Override
    public void close() {
        log.info("SendLogProducerInterceptor close");
    }
    //完成生产者配置时
    @Override
    public void configure(Map<String, ?> configs) {
        log.info("SendLogProducerInterceptor configure: {}", configs);
    }
}
